package com.caomt.rocketmq.config;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class SpringConfig {

//    @Value("$(rocketmq.consumer.groupName)")
    private String groupName = "test-demo";
//    @Value("$(rocketmq.consumer.namesrvAddr)")
    private String namesrvAddr = "192.168.138.212:9876;192.168.138.213:9876";
//    @Value("$(rocketmq.consumer.instanceName)")
    private String instanceName = "consumer.demo";
//    @Value("$(rocketmq.producer.sendMsgTimeout)")
    private int sendMsgTimeout = 1000;
//    @Value("$(rocketmq.producer.maxMessageSize)")
    private int maxMessageSize = 9999999;
//    @Value("$(rocketmq.producer.compressOver)")
    private int compressOver = 40000;
//    @Value("$(rocketmq.topic)")
    private String topic = "test-demo";
//    @Value("$(rocketmq.tag)")
    private String tag = "111";

    @Bean
    public DefaultMQProducer getRocketMQProducer(){
        DefaultMQProducer producer = new DefaultMQProducer(this.groupName);
        producer.setNamesrvAddr(this.namesrvAddr);
        producer.setInstanceName(this.instanceName);
        producer.setSendMsgTimeout(this.sendMsgTimeout);
        producer.setCompressMsgBodyOverHowmuch(this.compressOver);
        producer.setMaxMessageSize(this.maxMessageSize);

        try {
            producer.start();
        } catch (MQClientException e){
            e.printStackTrace();
        }
        return  producer;
    }

    @Bean
    public DefaultMQPushConsumer getRocketMQConsumer(){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.groupName);
        consumer.setNamesrvAddr(this.namesrvAddr);
        consumer.setInstanceName(this.instanceName);
        consumer.setConsumeMessageBatchMaxSize(1);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeConcurrentlyMaxSpan(2000);
        consumer.setPullThresholdForQueue(1000);
        consumer.setPullInterval(1000);
        consumer.setMessageModel(MessageModel.CLUSTERING);

        try {
            consumer.subscribe(this.topic, tag);
            consumer.registerMessageListener((MessageListenerConcurrently) (msgList, consumeConcurrentlyContext)->{
                try {
                    MessageExt msg = null;
                    for (MessageExt aMsgList : msgList) {
                        msg = aMsgList;
                        System.out.println("收到MQ消息：" + msg);
                    }
                } catch (JSONException e){
                    e.printStackTrace();
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            System.out.println("已启动Consumer【group：" + this.groupName +
                    ",instance:" + this.instanceName + "】，监听TOPIC-{" +
                    this.topic + "}, tag-{" + this.tag + "}");

        } catch (MQClientException e){
            e.printStackTrace();
        }
        return consumer;
    }
}
